/*!
 gfsc router module, gfs真正拓扑结构定义的实现与解析。
 * 
 *
 */
use anyhow::{anyhow, bail};
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use std::io::Write;
use std::time::UNIX_EPOCH;
use std::collections::HashMap;

#[cfg(feature = "tracelog")]
use tracing::{debug, error, info};
#[cfg(feature = "rotatelog")]
use log::{debug, error, info, warn};
use crate::*;

/**
定义获取writer，reader trait
 */
pub trait Router {
   fn get_writer(&self, dynasty_id: usize, ihash: i32) -> Result<String>;
   fn get_reader(&self, dynasty_id: usize, ihash: i32) -> Result<String>;    
}

/**
gfs中的节点，负责一段hashcode 【min,max】对应的文件存储，只能1个writer,可以n个reader
 */
#[derive(Debug, Serialize, Deserialize)]
pub struct Node {
    pub writer: String,
    pub readers: Vec<String>,
    pub max: i32,
    pub min: i32,
}
impl Node {
    fn is_here(&self, ihash: i32) -> bool {
        if ihash >= self.min && ihash <= self.max {
            return true;
        }
        false
    }
}
/**
Dynasty是一段时期内一个独立完整的gfs部署拓扑结构
 - Dynasty下可以由多个Node组成，
 - 每个Dynasty实体，都应该有一个独立的编号:dynasty, 是不同gfs部署的标识。
 - 通过部署不同Dynasty实体，可以完成历史数据(Dynasty)的逐步退出（只读），向新Dynasty的平滑迁移。
 - 也可以作为一种扩容的手段。
 */
#[derive(Debug, Serialize, Deserialize)]
pub struct Dynasty {
    pub nodes: Vec<Node>,
    pub dynasty: usize,
}
impl Dynasty {
    fn locate(&self, ihash: i32) -> Result<usize> {
        for i in 0..self.nodes.len() {
            if self.nodes[i].is_here(ihash) {
                return Ok(i);
            }
        }
        Err(anyhow!(
            "ihash = {} is not in here {:#?}",
            ihash,
            self.nodes
        ))
    }
    fn check(&self) -> Result<()>{
        let mut r:Vec<i32> = Vec::new();
        for i in 0..self.nodes.len(){
            let mut v = (self.nodes[i].min..(self.nodes[i].max+1)).map(|x|{x}).collect();
            r.append(&mut v);
        }
        if r.len() == 256 {
        return Ok(())
        }
        Err(anyhow!("error hash table, please check : {:#?}", r))
    }
}

/**
统筹多个Dynasty的拓扑结构关系，真正完成容灾备份和容量扩展的逻辑实现。
 - 一套完整的gfs部署，可以有多套Dynasty组成，但只能有一个Dynasty承担写入任务，其他都只能读、查、搜。
 - cur_dynasty表示当前承担写入工作的Dynasty标识。 
 - get_reader_url前，可以先判断 ：if is_preview_file(ext)
 - get_image_url前，可以先判断 ：if is_image_file(ext)
 */
#[derive(Debug, Serialize, Deserialize)]
pub struct GfsRouter {
    pub cur_dynasty: usize,
    pub dynasties: Vec<Dynasty>,
}

impl GfsRouter {
    pub fn new(conf :&str) -> Self {
        let router = gfsc_new_router(conf);
        match router {
            Ok(router) => router,
            Err( err)=> {
                panic!("{:#?}",err)
            }
        }
    }

    pub fn check(&self) -> Result<()> {
        if self.cur_dynasty >= self.dynasties.len() {
            return Err(anyhow!("cur_part to big {}! ", self.cur_dynasty))            
        }
        for j in 0..self.dynasties.len() {      
            if let Err(err) = self.dynasties[j].check() {
                return Err(err)
            }      
            for i in 0..256 {
                let r = self.dynasties[j].locate(i).unwrap_or(520);
                if r == 520 {
                    return Err(anyhow!("ihash = {} is not in route-table! ", i))                    
                }
            }
        }
        Ok(())
    }
    pub fn get_download_url2(&self, dynasty_id:usize, idx: i32, fname: &str) -> Result<String> {
        let ihash = gfs_filename_hash(fname);
        self.get_download_url(dynasty_id, idx, ihash, fname)        
    }
    pub fn get_reader_url2(&self, dynasty_id:usize, idx: i32, fname: &str) -> Result<String> {
        let ihash = gfs_filename_hash(fname);
        self.get_reader_url(dynasty_id, idx, ihash, fname)
    }
    pub fn get_image_url2(&self, dynasty_id:usize, idx: i32,fname: &str) -> Result<String> {        
        let ihash = gfs_filename_hash(fname);
        self.get_image_url(dynasty_id, idx, ihash, fname)       
    }
    pub fn get_download_url(&self, dynasty_id:usize, idx: i32, ihash:i32, fname: &str) -> Result<String> {   
        let token = new_token();     
        Ok(format!("{}/download?ihash={}&idx={}&name={}&token={}", self.get_reader(dynasty_id, ihash)?,ihash%256, idx, urlencoding::encode(fname), token))
    }
    pub fn get_master_download_url(&self, dynasty_id:usize, idx: i32, ihash:i32, fname: &str) -> Result<String> {        
        let token = new_token();     
        Ok(format!("{}/download?ihash={}&idx={}&name={}&token={}", self.get_writer(dynasty_id, ihash)?,ihash%256, idx, urlencoding::encode(fname), token))
    }   
    pub fn get_reader_url(&self, dynasty_id:usize, idx: i32, ihash:i32, fname: &str) -> Result<String> {        
        let token = new_token();     
        Ok(format!("{}/reader?ihash={}&idx={}&name={}&token={}", self.get_reader(dynasty_id, ihash)?,ihash%256, idx, urlencoding::encode(fname), token))
    }
    pub fn get_writer_url(&self, dynasty_id:usize, ihash:i32 ) -> Result<String> {        
        let token = new_token();     
        Ok(format!("{}/write/file/{}", self.get_writer(dynasty_id, ihash)?, token))
    }
    pub fn get_image_url(&self, dynasty_id:usize, idx: i32, ihash:i32, fname: &str) -> Result<String> {        
        let token = new_token();     
        Ok(format!("{}/image?ihash={}&idx={}&name={}&token={}", self.get_reader(dynasty_id, ihash)?,ihash%256, idx, urlencoding::encode(fname), token))
    }

    /**
     * 获取一个dynasty中每个node的一个reader的列表，多用于覆盖式的并行搜索或计算处理。
     */
    pub fn get_dynasty_readers(&self, dynasty_id: usize) -> Result< HashMap<(i32,i32),String> > {
        //let mut rets:Vec<String> = Vec::new();
        let mut readers = HashMap::new();        
        for i in 0..self.dynasties.len(){            
            if dynasty_id == self.dynasties[i].dynasty {
                for j in 0..self.dynasties[i].nodes.len() {
                    // let skey = format!("{}-{}", self.dynasties[i].nodes[j].min, self.dynasties[i].nodes[j].max);
                    let ikey = (self.dynasties[i].nodes[j].min, self.dynasties[i].nodes[j].max);
                    let reader = self.get_reader(dynasty_id, self.dynasties[i].nodes[j].min)?;
                    //rets.push(reader);
                    readers.insert(ikey, reader);
                }
            }
        }
        Ok(readers)
    }
        /**
     * 获取一个dynasty中每个node的一个writer的列表，多用于覆盖式的写入来部署文件。
     */
    pub fn get_dynasty_writers(&self, dynasty_id: usize) -> Result< HashMap<(i32,i32),String> > {
        let mut writers = HashMap::new();                
        for i in 0..self.dynasties.len(){            
            if dynasty_id == self.dynasties[i].dynasty {
                for j in 0..self.dynasties[i].nodes.len() {
                   // let skey = format!("{}-{}", self.dynasties[i].nodes[j].min, self.dynasties[i].nodes[j].max);
                   let ikey = (self.dynasties[i].nodes[j].min, self.dynasties[i].nodes[j].max);
                    let writer = self.get_writer(dynasty_id, self.dynasties[i].nodes[j].min)?;                    
                    writers.insert(ikey, writer);
                }
            }
        }
        Ok(writers)
    }
}

impl Router for Node {
    fn get_writer(&self, dynasty: usize, ihash: i32) -> Result<String> {
        debug!("here is not used {}", dynasty);
        if ihash >= self.min && ihash <= self.max {
            return Ok(self.writer.clone());
        }
        Err(anyhow!("ihash is not in [{},{})", self.min, self.max))
    }
    fn get_reader(&self, offset: usize, ihash: i32) -> Result<String> {
        debug!("offset for reader load-balance");
        if ihash >= self.min && ihash <= self.max {
            return Ok(self.readers[offset % self.readers.len()].clone());
        }
        Err(anyhow!("ihash is not in [{},{})", self.min, self.max))
    }
}
impl Router for Dynasty {
    fn get_writer(&self, dynasty: usize, ihash: i32) -> Result<String> {
        let idx = self.locate(ihash)?;
        return self.nodes[idx].get_writer(dynasty, ihash)        
    }
    fn get_reader(&self, offset: usize, ihash: i32) -> Result<String> {
        let idx = self.locate(ihash)?;
        return self.nodes[idx].get_reader(offset, ihash) 
    }
}
impl Router for GfsRouter {
    fn get_writer(&self, dynasty_id : usize, ihash: i32) -> Result<String> {
        if dynasty_id != self.cur_dynasty {
            error!(
                "you want to get writer from old Part-{}, cur_part = {}",
                dynasty_id, self.cur_dynasty
            );
        }
        if dynasty_id >= self.dynasties.len() {
            return Err(anyhow!("No such part: id = {}", dynasty_id));
        }
        self.dynasties[dynasty_id].get_writer(dynasty_id, ihash)
    }
    fn get_reader(&self, dynasty_id: usize, ihash: i32) -> Result<String> {
        if dynasty_id != self.cur_dynasty {
            error!(
                "you want to get writer from old Part-{}, cur_dynasty = {}",
                dynasty_id, self.cur_dynasty
            );
        }
        if dynasty_id >= self.dynasties.len() {
            return Err(anyhow!("No such part: id = {}", dynasty_id));
        }
        // 通过时钟，实现reader的轮转，实现load-balance
        let t = std::time::SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or(std::time::Duration::new(0, 0))
            .as_micros()
            % 32;
        self.dynasties[dynasty_id].get_reader(t as usize, ihash)
    }    
}
/**
 * 读取router.json的配置文件，生成gfs拓扑结构的路由对象。
 * 
 */
pub fn gfsc_new_router(conf_file :&str) -> Result< GfsRouter > {
    let src = String::from_utf8(std::fs::read(conf_file)?)?;
    let router: GfsRouter = serde_json::from_str(src.as_str())?;
    if let Err(err) = router.check(){
        println!("No: {}", err);
        bail!(err)   
    }
    Ok(router)
}

impl GfsRouter{
    /**
     * 生成默认的gfs拓扑结构定义router.json文件
     * 没有实际的生产意义，用于帮助开发者生成配置样本
     */
    pub fn try_generate(fname:&str)-> Self {
        let config_file = r##"
        {
            "cur_dynasty":1,        
            "dynasties":[
                {
                    "dynasty" : 0,
                    "nodes": [                    
                        {"max":255,"min":0, "writer":"http://localhost:9996","readers":["http://localhost:9996","http://localhost:9997"]} 
                        ]
                },
                {
                    "dynasty" : 1,
                    "nodes": [
                        {"max":127,"min":0, "writer":"http://localhost:7000","readers":["http://localhost:7000","http://localhost:7001"]},
                        {"max":255,"min":128, "writer":"http://localhost:7002","readers":["http://localhost:7002","http://localhost:7003"]} 
                        ]
                }
            ]      
        }            
        "##;
        
        let mut buffer = std::fs::File::create( fname ).unwrap();
        buffer.write_all(config_file.as_bytes()).unwrap();
        Self::new(fname )        
    }

    /**
     * 生成最简单的gfs拓扑结构
     * 没有实际的生产意义，用于帮助开发者本地调试
     */
    pub fn localhost_router(port: u16)-> Self {
        let router_src = r##"
        {
            "cur_dynasty":0,        
            "dynasties":[
                {
                    "dynasty" : 0,
                    "nodes": [                    
                        {"max":255,"min":0, "writer":"http://localhost:9996","readers":["http://localhost:9996"]} 
                        ]
                }
            ]      
        }            
        "##;
        let mut router: GfsRouter = serde_json::from_str(router_src).unwrap();
        router.dynasties[0].nodes[0].writer = format!("http://localhost:{}", port);
        router.dynasties[0].nodes[0].readers[0] = format!("http://localhost:{}", port);
        router
    }
}


#[test]
fn gfsc_config_router_test() {
    if let Err(err) = gfsc_config_router_ab() {
        println!("{}", err);
    }
}
fn gfsc_config_router_ab() -> Result<()> {
    let src = String::from_utf8(std::fs::read("./router.json")?)?;
    let router: GfsRouter = serde_json::from_str(src.as_str())?;
    println!("{:#?}", router);
    if let Err(err) = router.check(){
        println!("No: {}", err);
    }else{
        println!("Yes");
    }
    let writer = router.get_writer(router.cur_dynasty, 1)?;
    let reader1 = router.get_reader(router.cur_dynasty, 1)?;
    println!("writer = {}", writer);
    println!("reader1 = {}", reader1);
    let reader2 = router.get_reader(router.cur_dynasty, 1)?;
    println!("reader2 = {}", reader2);

    let reader2 = router.get_reader_url2(router.cur_dynasty, 1, "hello redis.conf")?;
    println!("reader url = {}", reader2);

    let reader2 = router.get_download_url2(router.cur_dynasty, 1, "hello redis.conf")?;
    println!("download url = {}", reader2);

    let readers = router.get_dynasty_readers(1)?;
    println!("{:#?}", readers);

    return Ok(());
}
